package com.lcy.struct.bTreePlus.disk.recovery.redo;

import com.lcy.struct.bTreePlus.disk.constants.BPlusPublicConstants;
import com.lcy.struct.bTreePlus.disk.file.FileUtil;
import lombok.SneakyThrows;

import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Arrays;
import java.util.List;

/**
 * @author lucongyang
 * @date Created in  10:27
 * @description
 */
class CommitRedo {
    /**
     * redo做完，提交redo中的事务
     *
     * @param transactionIds 事务id列表（不超过20位且全局唯一）
     */
    @SneakyThrows
    void commitRedo(Long... transactionIds) {
        List<Long> transactionIdList = Arrays.asList(transactionIds);
        FileChannel channel = RedoCenter.getInstance().getRedoChannel();
        long redoEndPosition = RedoCenter.getInstance().getRedoEndPosition();
        int redoStartStrLength = BPlusPublicConstants.REDO_START_STR.getBytes().length;
        int redoEndStrLength = BPlusPublicConstants.REDO_END_STR.getBytes().length;
        byte[] commitStrBytes = BPlusPublicConstants.REDO_END_COMMIT_STR.getBytes();
        long position = BPlusPublicConstants.HEAD_SIZE;
        boolean needCommit = false;
        int size = Math.max(redoStartStrLength + BPlusPublicConstants.MAX_TRANSACTION_ID_LENGTH, redoEndStrLength);
        while (position < redoEndPosition) {
            MappedByteBuffer mappedByteBuffer = channel.map(FileChannel.MapMode.READ_ONLY, position, size);
            byte[] bytes = new byte[size];
            mappedByteBuffer.get(bytes);
            String redoStartLengthStr = new String(bytes).trim();
            if (redoStartLengthStr.startsWith(BPlusPublicConstants.REDO_START_STR) && redoStartLengthStr.length() > redoStartStrLength) {
                long transactionId = Long.parseLong(redoStartLengthStr.substring(redoStartStrLength));
                if (transactionIdList.contains(transactionId)) {
                    needCommit = true;
                } else {
                    needCommit = false;
                }
            }

            if (redoStartLengthStr.startsWith(BPlusPublicConstants.REDO_END_STR)) {
                if (needCommit) {
                    FileUtil.getInstance().straightRewrite(channel, position + redoEndStrLength, commitStrBytes.length, true, commitStrBytes);
                }
                needCommit = false;
            }

            position += BPlusPublicConstants.REDO_PAGE_SIZE;
        }
    }

    static CommitRedo getInstance() {
        return Singleton.INSTANCE;
    }

    private static class Singleton {
        private static final CommitRedo INSTANCE = new CommitRedo();
    }

    private CommitRedo() {
    }
}
